package com.bw.fun;

import com.alibaba.fastjson.JSON;
import com.bw.bean.BoundSort;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class Mn6AsyncIOFunc extends RichAsyncFunction<BoundSort,BoundSort> {
    Connection conn = null;
    PreparedStatement ps1 = null;
    PreparedStatement ps2 = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
        ps1 = conn.prepareStatement("select * from tms01.dim_base_organ where id=?" );//机构表
        ps2 = conn.prepareStatement("select * from tms01.dim_base_region_info where id=?" ); //行政区域表
        conn.setAutoCommit(true);
    }
    @Override
    public void close() throws Exception {
        ps1.close();
        ps2.close();
        conn.close();
    }

    @Override
    public void asyncInvoke(BoundSort bs, ResultFuture<BoundSort> resultFuture) throws Exception {

        ps1.setString(1,bs.getOrg_id());
        ResultSet rs1 = ps1.executeQuery();
        if(rs1.next()){
            String org_name = JSON.parseObject(rs1.getString("line")).getString("org_name");
            bs.setOrg_name(org_name);
            String region_id = JSON.parseObject(rs1.getString("line")).getString("region_id");
            //查询城市
            ps2.setString(1,region_id.substring(0,4)+"00");
            ResultSet rs2 = ps2.executeQuery();
            if(rs2.next()){
                String cityName = JSON.parseObject(rs2.getString("line")).getString("name");
                bs.setCity_name(cityName);
            }
            //查询省份
            ps2.setString(1,region_id.substring(0,2)+"0000");
            ResultSet rs3 = ps2.executeQuery();
            if(rs3.next()){
                String provinceName = JSON.parseObject(rs3.getString("line")).getString("name");
                bs.setProvince_name(provinceName);
            }
        }
        resultFuture.complete(Collections.singleton(bs));
    }
}
